package com.xzx.flink.tableapi;

import com.xzx.flink.bean.ClickEvent;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/5/28 22:42
 */
public class TableAPI_03_Simple {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取数据源
        SingleOutputStreamOperator<ClickEvent> eventStream = env
                .fromElements(
                        new ClickEvent("Alice", "./home", 1000L),
                        new ClickEvent("Bob", "./cart", 1000L),
                        new ClickEvent("Alice", "./prod?id=1", 5 * 1000L),
                        new ClickEvent("Cary", "./home", 60 * 1000L),
                        new ClickEvent("Bob", "./prod?id=3", 90 * 1000L),
                        new ClickEvent("Alice", "./prod?id=7", 105 * 1000L)
                );
        // 获取表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 将数据流转换成表
        tableEnv.createTemporaryView("event_table", eventStream);
        // 查询 Alice 的访问 url 列表
        Table table1 = tableEnv.sqlQuery("select url,user from event_table where user='Alice'");
        tableEnv.toDataStream(table1).print("alice visit");

        // 统计每个用户的点击次数
        Table table2 = tableEnv.sqlQuery("select user,count(1) from event_table group by user");
        tableEnv.toChangelogStream(table2).print("count");

        env.execute();
    }
}
